package avicit;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class RocketMQConsumer {
    private static final String BROKER_NAME = "broker-a";
    private static final String TOPIC_NAME = "TestTopic";
    public static void main(String[] args) throws MQClientException {
        // 初始化消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TempConsumerGroup");

        // 设置NameServer地址
        consumer.setNamesrvAddr("10.216.38.233:8000");

        // 订阅Topic（与生产者一致）
        consumer.subscribe(TOPIC_NAME, "*");

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        // 打印消息内容
                        String messageBody = new String(msg.getBody(), "UTF-8");
                        System.out.printf("收到消息：%s | 内容：%s%n", msg, messageBody);
                    } catch (Exception e) {
                        e.printStackTrace();
                        // 处理失败时可以选择返回 RECONSUME_LATER 触发重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }

                // 返回消费成功状态
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();

        System.out.println("消费者已启动，等待消息...");
    }
}
